package com.bfxy.disruptor.common.producer;

import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

import com.bfxy.disruptor.common.DataProvider;
import com.bfxy.disruptor.common.event.EventPoller;
import com.bfxy.disruptor.common.Util;
import com.bfxy.disruptor.common.balance.ProcessingSequenceBarrier;
import com.bfxy.disruptor.common.balance.Sequence;
import com.bfxy.disruptor.common.balance.SequenceBarrier;
import com.bfxy.disruptor.common.balance.SequenceGroups;
import com.bfxy.disruptor.common.waitStraty.WaitStrategy;
//单生产者或者多生产者的父类。
public abstract class AbstractProductor implements Productor {
	private static final AtomicReferenceFieldUpdater<AbstractProductor, Sequence[]> SEQUENCE_UPDATER = AtomicReferenceFieldUpdater
			.newUpdater(AbstractProductor.class, Sequence[].class, "gatingSequences");//对gatingSequences进行原子更新

	protected final int bufferSize;
	protected final WaitStrategy waitStrategy;// 等待策略
	protected final Sequence cursor = new Sequence(Productor.INITIAL_CURSOR_VALUE);//生产者投递数据时候投递的位置，会有线程安全。这个实例只有一个，就是生产者的位置，保存在单或多生产者里面。初始值-1.
	protected volatile Sequence[] gatingSequences = new Sequence[0];//消费者消费到哪个位置

	public AbstractProductor(int bufferSize, WaitStrategy waitStrategy) {
		if (bufferSize < 1) {
			throw new IllegalArgumentException("bufferSize must not be less than 1");
		}
		if (Integer.bitCount(bufferSize) != 1) {
			throw new IllegalArgumentException("bufferSize must be a power of 2");
		}

		this.bufferSize = bufferSize;
		this.waitStrategy = waitStrategy;
	}

	@Override
	public final long getCursor() {
		return cursor.get();
	}

	@Override
	public final int getBufferSize() {
		return bufferSize;
	}

	@Override
	public final void addGatingSequences(Sequence... gatingSequences) {
		SequenceGroups.addSequences(this, SEQUENCE_UPDATER, this, gatingSequences);
	}

	@Override
	public boolean removeGatingSequence(Sequence sequence) {
		return SequenceGroups.removeSequence(this, SEQUENCE_UPDATER, sequence);
	}

	@Override
	public long getMinimumSequence() {
		return Util.getMinimumSequence(gatingSequences, cursor.get());
	}

	//disruptor里面创建消费者处理器时候，会创建栅栏。然后调用ringBuffer来创建栅栏。ringBuffer又调用里面的生产者处理器来创建栅栏。
	@Override
	public SequenceBarrier newBarrier(Sequence... sequencesToTrack) {
		//this=生产者（这个生产者只有一个），cursor=-1，sequencesToTrack=new Sequence[0]是一个长度为0的数组。
		//生产者没有栅栏，只有消费者才有栅栏，并且栅栏里面也有单或者多生产者sequencer。
		return new ProcessingSequenceBarrier(this, waitStrategy, cursor, sequencesToTrack);
	}

	@Override
	public <T> EventPoller<T> newPoller(DataProvider<T> dataProvider, Sequence... gatingSequences) {
		return EventPoller.newInstance(dataProvider, this, new Sequence(), cursor, gatingSequences);
	}
}